package net.bwie.jtp.dws.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

public class JtpTrafficAppStartMinuteWindowDwsJob {
    public static void main(String[] args) {

        // TODO: 2025/5/21 创建执行环境
        TableEnvironment tabEnv = getTableEnv();
        // TODO: 2025/5/21 输入表-input:映射到kafka消息队列
        createInputTable(tabEnv);
        // TODO: 2025/5/21 数据处理
        handle(tabEnv);
        // TODO: 2025/5/21 输出表-output:映射到CK表
        createOutputTable(tabEnv);

        LookUp(tabEnv);

        // TODO: 2025/5/21 保存数据
        saveToClickHouse(tabEnv);




    }

    private static void LookUp(TableEnvironment tabEnv) {
        // TODO: 2025/5/23 维表;lookup join
        Table opentable = tabEnv.sqlQuery(
                "select\n" +
                        "    window_start as window_start_time,\n" +
                        "    window_end as window_end_time,\n" +
                        "    open_ad_id,\n" +
                        "    count(open_ad_id) as open_ad_id_count,\n" +
                        "    sum(open_ad_ms)                              as open_ad_ms,\n" +
                        "    sum(open_ad_skip_ms)                         as open_ad_skip_ms,\n" +
                        "    unix_timestamp() * 1000                      as ts,\n" +
                        "    proctime() as proc_time\n" +
                        "from\n" +
                        "    TABLE(\n" +
                        "        TUMBLE(TABLE start_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES)\n" +
                        "    )t1\n" +
                        "    GROUP BY window_start, window_end,open_ad_id"
        );
        tabEnv.createTemporaryView("start_ad_report_table", opentable);

        // TODO: 2025/5/23 维表关联
        tabEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS ads_info_mysql_source_dim\n" +
                        "(\n" +
                        "    id               INT  comment '主键id',\n" +
                        "    name             String  COMMENT '广告名称',\n" +
                        "    content          String           COMMENT '广告内容（图片URL/视频URL/H5代码）',\n" +
                        "    type             String   COMMENT '广告类型(IMAGE/VIDEO/HTML5)',\n" +
                        "    start_time       TIMESTAMP      COMMENT '展示开始时间',\n" +
                        "    end_time         TIMESTAMP      COMMENT '展示结束时间',\n" +
                        "    is_active        INT         COMMENT '是否启用',\n" +
                        "    target_condition String COMMENT '定向条件（JSON格式）',\n" +
                        "    created_at       TIMESTAMP  comment '数据插入时间',\n" +
                        "    updated_at       TIMESTAMP  comment '数据更新时间',\n" +
                        "    primary key (id) not enforced\n" +
                        ")with(\n" +
                        "    'connector' = 'jdbc',\n" +
                        "    'url' = 'jdbc:mysql://node103:3306/jtp_ads',\n" +
                        "    'table-name' = 'ads_info',\n" +
                        "    'lookup.cache.max-rows' = '10',\n" +
                        "    'lookup.cache.ttl' = '1 hour',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'driver' = 'com.mysql.cj.jdbc.Driver'\n" +
                        ")"
        );

        Table table = tabEnv.sqlQuery(

                "select\n" +
                        "    a1.window_start_time,\n" +
                        "    a1.window_end_time,\n" +
                        "    a2.name as open_ad_id,\n" +
                        "    a1.open_ad_id_count,\n" +
                        "    a1.open_ad_ms,\n" +
                        "    a1.open_ad_skip_ms,\n" +
                        "    a1.ts\n" +
                        "from start_ad_report_table as a1\n" +
                        "left join ads_info_mysql_source_dim\n" +
                        " FOR system_time as of proc_time as a2\n" +
                        "on  cast(a1.open_ad_id as int)  = a2.id"
        );
        tabEnv.createTemporaryView("join_table",table);

    }


    private static void createOutputTable(TableEnvironment tabEnv) {
        // TODO: 2025/5/21 启动入口统计表
        tabEnv.executeSql(
                "create table dws_traffic_app_start_window_report_clickhouse_sink (\n" +
                        "    window_start_time String comment '计算窗口开始时间',\n" +
                        "    window_end_time   String comment '计算窗口结束时间',\n" +
                        "    entry             String comment '启动入口',\n" +
                        "    entry_count       bigint comment '启动次数',\n" +
                        "    loading_time      bigint comment '页面加载时间',\n" +
                        "    ts                bigint comment '数据插入时间戳'\n" +
                        ")with(\n" +
                        "    'connector' = 'clickhouse',\n" +
                        "    'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "    'table' = 'dws_traffic_app_start_window_report',\n" +
                        "    'username' = 'default',\n" +
                        "    'password' = '',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

        // TODO: 2025/5/21 启动日志广告统计表
        tabEnv.executeSql(
                "create table dws_traffic_app_start_ad_window_report_clickhouse_sink (\n" +
                        "  window_start_time String comment '计算窗口开始时间',\n" +
                        "  window_end_time   String comment '计算窗口结束时间',\n" +
                        "  open_ad_id        String comment '广告id',        \n" +
                        "  open_ad_id_count   bigint comment '启动次数',    \n" +
                        "  open_ad_ms      bigint comment '广告总共播放时间',\n" +
                        "  open_ad_skip_ms      bigint comment '用户跳过广告的时间',\n" +
                        "  ts                bigint comment '数据插入时间戳'\n" +
                        ")with(\n" +
                        "    'connector' = 'clickhouse',\n" +
                        "    'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "    'table' = 'dws_traffic_app_start_ad_window_report',\n" +
                        "    'username' = 'default',\n" +
                        "    'password' = '',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

    }

    private static void handle(TableEnvironment tabEnv) {
        // TODO: 2025/5/21 提取启动日志字段
        Table startLogTable = tabEnv.sqlQuery(
                "select\n" +
                        "    `start`['entry'] as entry,\n" +
                        "    CAST(`start`['loading_time'] as BIGINT) as loading_time,\n" +
                        "    `start`['open_ad_id'] as open_ad_id,\n" +
                        "    CAST(`start`['open_ad_ms'] as BIGINT) as open_ad_ms,\n" +
                        "    cast(`start`['open_ad_skip_ms'] as BIGINT)as open_ad_skip_ms,\n" +
                        "    row_time\n"+
                        "from dwd_traffic_start_log_kafka_source\n" +
                        "where `start`['entry'] is not null "
        );
        tabEnv.createTemporaryView("start_log_table", startLogTable);

        // TODO: 2025/5/21 启动日志处理根据类型统计设置窗口进行分组聚合计算
//        Table startReportTable = tabEnv.sqlQuery(
//                "select \n" +
//                        "    tumble_start(row_time, interval '1' minutes) as window_start_time,\n" +
//                        "    tumble_end(row_time, interval '1' minutes) as window_end_time,\n" +
//                        "    entry,\n" +
//                        "    count(entry) as entry_count,\n" +
//                        "    sum(loading_time) as loading_time,\n" +
//                        "    unix_timestamp() * 1000 as ts\n" +
//                        "from start_log_table\n" +
//                        "group by tumble(row_time, interval '1' minutes),entry"
//        );
//        tabEnv.createTemporaryView("start_report_table", startReportTable);
        // TODO: 2025/5/23 表值函数

        Table startTable = tabEnv.sqlQuery(
                "select window_start as window_start_time,\n" +
                        "       window_end   as window_end_time,\n" +
                        "       entry,\n" +
                        "       count(entry) as entry_count,\n" +
                        "       sum(loading_time) as loading_time,\n" +
                        "       unix_timestamp() * 1000 as ts\n" +
                        "from table(\n" +
                        "        TUMBLE(TABLE start_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))\n" +
                        "GROUP BY window_start, window_end,entry"
        );
       tabEnv.createTemporaryView("start_report_table", startTable);


        // TODO: 2025/5/21 启动广告日志处理根据类型统计设置窗口进行分组聚合计算
//        Table startAdReportTable = tabEnv.sqlQuery(
//                "select\n" +
//                        "    tumble_start(row_time, interval '1' minutes) as window_start_time,\n" +
//                        "    tumble_end(row_time, interval '1' minutes) as window_end_time,\n" +
//                        "    open_ad_id,\n" +
//                        "    count(open_ad_id) as open_ad_id_count,\n" +
//                        "    sum(open_ad_ms) as open_ad_ms,\n" +
//                        "    sum(open_ad_skip_ms) as open_ad_skip_ms,\n" +
//                        "    unix_timestamp() * 1000 as ts\n" +
//                        "from start_log_table\n" +
//                        "group by tumble(row_time, interval '1' minutes),open_ad_id"
//        );
//        tabEnv.createTemporaryView("start_ad_report_table", startAdReportTable);
        // TODO: 2025/5/23 表值函数
//        Table adReportTable = tabEnv.sqlQuery(
//                "select \n" +
//                        "    window_start as window_start_time,\n" +
//                        "    window_end as window_end_time,\n" +
//                        "    open_ad_id,\n" +
//                        "    count(open_ad_id) as open_ad_id_count,\n" +
//                        "    sum(open_ad_ms)                              as open_ad_ms,\n" +
//                        "    sum(open_ad_skip_ms)                         as open_ad_skip_ms,\n" +
//                        "    unix_timestamp() * 1000                      as ts\n" +
//                        "from \n" +
//                        "    TABLE(\n" +
//                        "        TUMBLE(TABLE start_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES)\n" +
//                        "    )\n" +
//                        "    GROUP BY window_start, window_end,open_ad_id"
//        );
//        tabEnv.createTemporaryView( "start_ad_report_table", adReportTable);


    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE dwd_traffic_start_log_kafka_source (\n" +
                        "  `common` MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
                        "  `start` MAP<STRING, STRING> COMMENT '启动信息',\n" +
                        "  `ts` BIGINT,\n" +
                        "  row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss.SSS')),\n" +
                        "  WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE\n" +
                        ") WITH (\n" +
                        "   'connector' = 'kafka',\n" +
                        "   'topic' = 'dwd-traffic-start-log',\n" +
                        "   'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "   'properties.group.id' = 'gid_dws_log_app_start',\n" +
                        "   'scan.startup.mode' = 'earliest-offset',\n" +
                        "   'format' = 'json',\n" +
                        "   'json.fail-on-missing-field' = 'false',\n" +
                        "   'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );
    }

    private static TableEnvironment getTableEnv() {
        // 创建执行环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useAnyPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 设置时区
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 返回对象
        return tabEnv;
    }


    private static void saveToClickHouse(TableEnvironment tabEnv) {

        // TODO: 2025/5/21 启动入口统计表插入数据

        tabEnv.executeSql(
                "insert into dws_traffic_app_start_window_report_clickhouse_sink\n" +
                        "select\n" +
                        "    DATE_FORMAT(window_start_time,  'yyyy-MM-dd HH:mm:ss') as window_start_time,\n" +
                        "    DATE_FORMAT(window_end_time,  'yyyy-MM-dd HH:mm:ss') as window_end_time,\n" +
                        "    entry,\n" +
                        "    entry_count,\n" +
                        "    loading_time,\n" +
                        "    ts\n" +
                        "from start_report_table"
        );
        // TODO: 2025/5/21 启动日志广告统计表插入数据
        tabEnv.executeSql(
                "\n" +
                        "insert into dws_traffic_app_start_ad_window_report_clickhouse_sink\n" +
                        "select\n" +
                        "    DATE_FORMAT(window_start_time,  'yyyy-MM-dd HH:mm:ss') as window_start_time,\n" +
                        "    DATE_FORMAT(window_end_time,  'yyyy-MM-dd HH:mm:ss') as window_end_time,\n" +
                        "    open_ad_id,\n" +
                        "    open_ad_id_count,\n" +
                        "    open_ad_ms,\n" +
                        "    open_ad_skip_ms,\n" +
                        "    ts\n" +
                        "from join_table"
        );

        TableResult tableResult = tabEnv.executeSql(
                "\n" +
                        "insert into dws_traffic_app_start_ad_window_report\n" +
                        "select\n" +
                        "    DATE_FORMAT(window_start_time,  'yyyy-MM-dd HH:mm:ss') as window_start_time,\n" +
                        "    DATE_FORMAT(window_end_time,  'yyyy-MM-dd HH:mm:ss') as window_end_time,\n" +
                        "    open_ad_id,\n" +
                        "    open_ad_id_count,\n" +
                        "    open_ad_ms,\n" +
                        "    open_ad_skip_ms,\n" +
                        "    ts\n" +
                        "from join_table"
        );
        tableResult.print();
    }

}
